import numpy as np
import pandas as pd
from es_pandas import es_pandas

from air_web.data_platform import sql_engine


class GetConcall:
    AREA_MAP = {"on5": "shi", "on7": "xian", "org_no": "org_name"}

    def __init__(self, es_host, index_name, tg_csv):
        self.ep = es_pandas(es_host)
        self.index_name = index_name
        self.tg_csv = tg_csv

        self.e_data_table = "e_data_mp"
        self.zb_table = "e_data_mp_zb"

    def proc_data(self, data_df, org_df):
        data_df.loc[data_df["on7"] == "13401", "on7"] = "1340199"
        data_df["on5"] = data_df["on7"].str[0:5]
        data_df = data_df.astype(
            {
                "mp_level": "int64",
                "org_no": "int64",
                "on7": "int64",
                "on5": "int64",
            },
            errors="ignore",
        )

        for area_code, area_name in self.AREA_MAP.items():
            temp_org_df = org_df.rename(columns={"org_no": area_code})
            data_df = data_df.merge(
                temp_org_df[[area_code, "org_name"]], on=area_code, how="left"
            )
            data_df.rename(columns={"org_name": area_name}, inplace=True)

        sql_engine.insert_df(data_df, "c_cons_all", drop_id=False)
        data_df["mp_level"] = data_df["mp_level"].replace({None: np.nan})
        self.ep.to_es(data_df, self.index_name, request_timeout=60)

    def get_zb_data(self):
        sql = """select edm.cons_no, 
                        cons_name, 
                        meter_id,
                        mp_id, 
                        mp_level, 
                        id,
                        org_no, 
                        area_no as on7, 
                        ctm.c_id, 
                        tm.type_id, 
                        tm.type_code, 
                        tm.type_code_sort,
                        tm.pare_type_id,
                        tm.pare_type_code
                 from {} edm
                 INNER JOIN
                 (SELECT c_id, cons_no, type_id
                  FROM cons_type_map
                 )ctm on ctm.cons_no = edm.cons_no
                 INNER JOIN
                 (SELECT pare_type_id, pare_type_code, type_id, type_code, type_code_sort
                  FROM type_map
                 )tm on tm.type_id=ctm.type_id""".format(
            self.zb_table
        )
        data_df = sql_engine.query(sql)
        return data_df

    def get_gb_data(self):
        sql = """select cons_no, 
                        cons_name, 
                        meter_id,
                        mp_id, 
                        mp_level, 
                        id,
                        org_no, 
                        area_no as on7, 
                        tg_id,
                        (@row_number := @row_number + 1) as c_id,
                        '城市台区' as pare_type_code,
                        201 as pare_type_id, 
                        '城市居民' as type_code, 
                        20101 as type_id, 
                        2 as type_code_sort
                 from {}, (SELECT @row_number := 300000) as t
                 where cons_sort = '06'
                   """.format(
            self.e_data_table
        )
        data_df = sql_engine.query(sql)
        return data_df

    def get_all_data(self):
        # cons_sort 01/02:专变 06:公变
        sql = """select edm.cons_no, 
                       cons_name, 
                       meter_id,
                       mp_id, 
                       mp_level, 
                       id,
                       org_no, 
                       area_no as on7, 
                       ctm.c_id, 
                       tm.type_id, 
                       tm.type_code, 
                       tm.type_code_sort,
                       tm.pare_type_id,
                       tm.pare_type_code
                from {} edm
                INNER JOIN
                (SELECT c_id, cons_no, type_id
                 FROM cons_type_map
                )ctm on ctm.cons_no = edm.cons_no
                INNER JOIN
                (SELECT pare_type_id, pare_type_code, type_id, type_code, type_code_sort
                 FROM type_map
                )tm on tm.type_id=ctm.type_id
                where cons_sort in ('01', '02', '06')""".format(
            self.e_data_table
        )
        data_df = sql_engine.query(sql)
        return data_df

    def get_tg_info(self):
        tg_df = pd.read_csv(self.tg_csv, header=0, dtype=str)
        tg_id_list = tg_df["TG_ID"].tolist()
        return tg_id_list

    def get_real_org_no(self):
        sql = "select org_no, org_name, org_level from real_org_no"
        org_df = sql_engine.query(sql)
        return org_df

    def get_on7_list(self, table_name, where_sql=""):
        sql = "select distinct area_no as on7 from {} {}".format(
            table_name, where_sql
        )
        on7_df = sql_engine.query(sql)
        on7_list = on7_df["on7"].tolist()
        return on7_list

    def main(self):
        org_df = self.get_real_org_no()

        df = self.get_all_data()
        print(len(df))
        self.proc_data(df, org_df)

    def create_map(self, cons_type_csv):
        # 专变
        df = pd.read_csv(cons_type_csv, header=0, dtype=str)
        df.rename(
            columns={
                "户号": "cons_no",
                "十一大行业": "pare_type_code",
                "133小行业": "type_code",
            },
            inplace=True,
        )

        type_map_df = df[["pare_type_code", "type_code"]].drop_duplicates()
        type_map_df["pare_type_id"] = (
            type_map_df["pare_type_code"].factorize()[0] + 101
        )
        res_type_map_df = pd.DataFrame()
        for para_type_id, group_df in type_map_df.groupby("pare_type_id"):
            group_df["type_id"] = (
                group_df["type_code"].factorize()[0] + para_type_id * 100 + 1
            )
            res_type_map_df = pd.concat([res_type_map_df, group_df])
            res_type_map_df["type_code_sort"] = 1
        sql_engine.insert_df(res_type_map_df, "type_map")

        df = df.merge(
            res_type_map_df[["pare_type_code", "type_code", "type_id"]],
            on=["pare_type_code", "type_code"],
            how="left",
        )
        cons_type_df = df[["cons_no", "type_id"]].drop_duplicates()
        sql_engine.insert_df(cons_type_df, "cons_type_map")

        # 公变
        gb_type_df = pd.DataFrame(
            {
                "pare_type_id": ["201"],
                "pare_type_code": ["城市台区"],
                "type_id": ["20101"],
                "type_code": ["城市居民"],
                "type_code_sort": ["2"],
            }
        )
        sql_engine.insert_df(gb_type_df, "type_map")

        sql = """select cons_no, 
                        tg_id,
                        20101 as type_id
                 from {}
                 where cons_sort = '06'""".format(
            self.e_data_table
        )
        gb_cons_df = sql_engine.query(sql)
        tg_id_list = self.get_tg_info()
        gb_cons_df = gb_cons_df.loc[gb_cons_df["tg_id"].isin(tg_id_list)]
        cons_type_df = gb_cons_df[["cons_no", "type_id"]].drop_duplicates()
        sql_engine.insert_df(cons_type_df, "cons_type_map")

    def e_data_zb(self):
        # zb_where_sql = "where cons_sort in ('01', '02')"
        # zb_on7_list = self.get_on7_list(self.zb_table, zb_where_sql)
        # for on7 in zb_on7_list:
        sql = """select cons_no, 
                        cons_name, 
                        meter_id,
                        mp_id, 
                        mp_level, 
                        id,
                        org_no, 
                        area_no
                 from {}
                 where cons_sort in ('01', '02')""".format(
            self.e_data_table
        )
        data_df = sql_engine.query(sql)
        sql_engine.insert_df(data_df, "e_data_mp_zb", drop_id=False)


if __name__ == "__main__":
    es_host = "zxtech:Zxod112_shining10@192.168.80.224:19200"
    index_name = "hbfh_c_cons_all_idx"
    tg_csv = "/home/smxu/hebei/file/gbtg.csv"
    cons_type_csv = "/home/smxu/hebei/file/zb_cons.csv"
    gc = GetConcall(es_host, index_name, tg_csv)
    # gc.e_data_zb()
    gc.create_map(cons_type_csv)
    gc.main()
